package akka.streams

import akka.actor.ActorSystem
import akka.stream.scaladsl.{Flow, Framing, Sink, Tcp}
import akka.stream.{ActorMaterializer, Materializer}
import akka.util.ByteString

import scala.util.{Failure, Success}

/**
  * Created by shichen on 2017/7/10.
  * https://gist.github.com/sschaef/bd5ee6273ddaa7b015af#file-simpletcpserver-scala
  */
object StreamTcpServer extends App{
  val addr = "127.0.0.1"
  val port = 6666

  val serverLogic = Flow[ByteString]
    .via(Framing.delimiter(ByteString("\n"),maximumFrameLength = 256,allowTruncation = true))
    .map(_.utf8String)
    .map(msg => s"Server responds to $msg")
    .map(ByteString(_))

  def mkServer(addr: String, port: Int)(implicit system:ActorSystem,materializer: Materializer): Unit = {
    import system.dispatcher
    val connectionHandler = Sink.foreach[Tcp.IncomingConnection] { conn =>
      println(s"incoming connection from ${conn.remoteAddress}")
      conn.handleWith(serverLogic)//handler is Flow
    }
    val incommingConnections = Tcp().bind(addr,port)//异步启动，只有future完成了才能接收连接
    val binding = incommingConnections.to(connectionHandler).run()
    binding.onComplete {
      case Success(b) =>
        println(s"server started,listening on :${b.localAddress}")
      case Failure(e) =>
        println(s"server could not be bound to $addr:$port: ${e.getMessage}")
    }
  }

  def mkAkkaServer(): Unit = {
    implicit val server = ActorSystem("server")
    implicit val materializer = ActorMaterializer()
    mkServer(addr,port)
  }

  mkAkkaServer()
}
